package flink_p1

import org.apache.flink.api.common.functions.ReduceFunction
import org.apache.flink.api.scala.createTypeInformation
import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

/**
 *
 *
 *
 * flink中的时间语义：
 *    1. ProcessingTime:   当数据进入窗口的系统时间
 *       2. Event Time:        当前数据在数据源中产生的时间
 *       3. Ingestion Time:    数据进入到flink source的系统时间
 *
 * */

object FlinkTest_19_window_eventTime {


  /**
   * 使用event time来计算WordCount
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)


    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)
      .assignAscendingTimestamps(_.split(" ")(0).toLong)

    // assignAscendingTimestamps 从数据中取出Event Time


    socketStream
      .map(data => {
        (data.split(" ")(1), 1)
      }).keyBy(_._1)
      .timeWindow(Time.seconds(3))
      .reduce(new ReduceFunction[(String, Int)] {
        override def reduce(value1: (String, Int), value2: (String, Int)): (String, Int) = {

          (value1._1, value2._2 + value2._2)
        }
      }, new WindowFunction[(String, Int), (String, Int), String, TimeWindow] {
        override def apply(key: String, window: TimeWindow, input: Iterable[(String, Int)], out: Collector[(String, Int)]): Unit = {

          println(s"start: ${window.getStart}")
          println(s"end: ${window.getEnd}")
        }
      }


      ).print()


    env.execute()
  }

}
